package cn.donghuapian.consumer;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @description:
 * @author: hzw
 * @time: 2024/12/16 16:47
 */
@Slf4j
@Component
public class CircuitBreakerConsumer {


    @RabbitListener(queuesToDeclare = @Queue("aurelionSol"))
    public void process(String msg, Channel channel, Message message) {
        try {

            testCircuitBreaker(msg);
            System.out.println("调用结束" + msg);
            msgAck(msg, channel, message);
        } catch (Exception e) {
            System.out.println("异常继续" + msg);
        }
    }

    @CircuitBreaker(name = "breakName", fallbackMethod = "fallback")
    public void testCircuitBreaker(String message) throws InterruptedException {
        int i = Integer.parseInt(message);
        Thread.sleep(5 * 1000L);
//        while(i > 2) {
//            throw new RuntimeException("主动异常");
//        }
        System.out.println("进入熔断方法" + message);
    }

    private void fallback(String message, Throwable throwable) {
        // 处理失败情况的逻辑
        System.err.println("异常回调");

    }

    private void fallback(String message, CallNotPermittedException exception) {
        // 处理失败情况的逻辑
        System.err.println("熔断回调");

    }

    //消息统一ACK
    private void msgAck(String msg, Channel channel, Message message) {
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.info("消息处理失败：{}", msg);
            //确认消息
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                //channel对象重新发送消息到队尾
                channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
                        message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
                        JSON.toJSONBytes(msg));
                log.info("消息放到尾部：{}", msg);
            } catch (IOException ioException) {
                log.info("消息放到尾部失败：{}" + msg);
            }
        }
    }
}
